package com.business.lgh.naocs.naming.consistency;

import com.business.lgh.naocs.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl;
import com.business.lgh.naocs.naming.core.ServerMemberManager;
import com.business.lgh.naocs.naming.misc.Datum;
import com.business.lgh.naocs.naming.misc.NamingProxy;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author lgh
 * @version V1.0
 * @Package com.business.lgh.naocs.naming.consistency
 * @date 2020/6/30 12:59 上午
 */
@Component
public class TaskDispatcher implements InitializingBean {

    /**
     * 任务队列
     */
    private BlockingQueue<String> taskQueue=new LinkedBlockingQueue();

    @Autowired
    DistroConsistencyServiceImpl distroConsistencyService;

    @Autowired
    ServerMemberManager serverMemberManager;


    @Override
    public void afterPropertiesSet() throws Exception {
        GlobalExecutorPools.submitNow(new TaskProcessor());
    }

    class TaskProcessor implements Runnable{

        ArrayList<String> keys;
        int dataCount=0;
        long lastSyncTime=System.currentTimeMillis();

        @Override
        public void run() {
            try {
                while (true) {
                    String key = taskQueue.poll();
                    if (dataCount==0){
                        keys=new ArrayList<>();
                    }
                    keys.add(key);
                    dataCount++;
                    if (dataCount>10||System.currentTimeMillis()-lastSyncTime>30){
                        lastSyncTime=System.currentTimeMillis();
                        Map<String, Datum> tmpDataStore=new HashMap<>();
                        for (String syncKey:keys){
                            tmpDataStore.put(syncKey,distroConsistencyService.get(syncKey));
                        }
                        for (String address:serverMemberManager.getServerList().keySet()){
                            if (address.equals(serverMemberManager.getLocalAddress()))
                                continue;
                            NamingProxy.syncData(address,tmpDataStore);
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
